package top.whysu.edevice.emqx;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.commons.lang3.StringUtils;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallbackExtended;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import top.whysu.edevice.constant.EmqConstant;
import top.whysu.edevice.dto.emqx.MqttBaseRequest;
import top.whysu.edevice.dto.emqx.MqttBaseResponse;
import top.whysu.edevice.global.ThreadPoolBuilder;
import top.whysu.edevice.service.device.DeviceEmqService;
import top.whysu.edevice.utils.Coder;
import top.whysu.edevice.utils.DeviceJoinUtil;
import top.whysu.edevice.utils.SpringUtil;

/**
 * 设备接入EMQ回调处理类
 */
public class DeviceSubcribeCallBack implements MqttCallbackExtended {

    private static final Logger LOGGER = LoggerFactory.getLogger(DeviceSubcribeCallBack.class);

    /**
     * 设备接入处理service
     */
    private DeviceEmqService deviceEmqService = SpringUtil.getBean(DeviceEmqService.class);
    private ThreadPoolBuilder threadPoolBuilder = SpringUtil.getBean(ThreadPoolBuilder.class);
    DeviceMqttClient deviceMqttClient = DeviceMqttClient.getInstance();

    @Override
    public void connectComplete(boolean reconnect, String serverUri) {
        LOGGER.info("社区中心EMQ连接完成");
        try {
            deviceMqttClient.getClient().subscribe(EmqConstant.Topic.CLIENT_STATUS, 0);
        } catch (MqttException e) {
            LOGGER.error("设备上下线主题订阅失败！", e);
        }
    }

    @Override
    public void connectionLost(Throwable throwable) {
        LOGGER.error("社区中心EMQ连接丢失", throwable);
    }

    @Override
    public void messageArrived(String topic, MqttMessage mqttMessage) {
        try {
            if (topic.contains(EmqConstant.Topic.CLIENT_CONNECT_PREFIX)) {
                // 上下线消息
                threadPoolBuilder.getThreadPool(ThreadPoolBuilder.THREADPOOL_DEVICE_KEY_ENTRANCE_ON_LINE).execute(() -> {
                    try {
                        // TODO
                    } catch (Exception e) {
                        LOGGER.error("接口设备上下线消息异常", e);
                    }
                });
            } else {
                // 设备请求或应答数据
                String msg = deviceMsgDecrypt(topic, mqttMessage);
                // 消息流打印
                LOGGER.info("接收设备消息-> topic = {} \npayload ={} ", topic, msg);
                // 消息是否为空，为空直接return
                if (StringUtils.isBlank(msg)) {
                    LOGGER.info("消息体为空！");
                } else if (DeviceJoinUtil.isResponse(msg)) {
                    // 应答数据处理：判断是否为应答消息，如果为应答消息，则直接将消息设置到锁参数中，并解锁即可。
                    threadPoolBuilder.getThreadPool(ThreadPoolBuilder.THREADPOOL_DEVICE_KEY_ENTRANCE_RESPONSE).execute(() -> {
                        DeviceJoinUtil.deviceResponseMsgHandler(msg);
                    });
                } else {
                    // 请求数据处理
                    MqttBaseRequest mqttBaseRequest = JSON.parseObject(msg, MqttBaseRequest.class);
                    switch (mqttBaseRequest.getMethod()) {
                        // APP登录
                        case EmqConstant.Method.LOGIN:
                            threadPoolBuilder.getThreadPool(ThreadPoolBuilder.THREADPOOL_APP_LOGIN).submit(() -> {
                                responseDevice(topic, mqttBaseRequest, deviceEmqService.appLogin(mqttBaseRequest));
                            });
                            break;
                        // 报警信息请求
                        case EmqConstant.Method.ALARM_INFO:
                            threadPoolBuilder.getThreadPool(ThreadPoolBuilder.THREADPOOL_ALARM_INFO).submit(() -> {
                                responseDevice(topic, mqttBaseRequest, deviceEmqService.alarmInfo(mqttBaseRequest));
                            });
                            break;
                        // 添加摄像头/船舶请求
                        case EmqConstant.Method.ADD_DEVICE:
                            threadPoolBuilder.getThreadPool(ThreadPoolBuilder.THREADPOOL_ADD_DEVICE).submit(() -> {
                                responseDevice(topic, mqttBaseRequest, deviceEmqService.addDevice(mqttBaseRequest));
                            });
                            break;
                        default:
                            break;
                    }
                }
            }
        } catch (Exception e) {
            LOGGER.error("接收设备消息异常", e);
        }
    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {

    }

    /**
     * @param topic       主题
     * @param mqttMessage 消息体
     * @return 解密处理后的消息体
     */
    private String deviceMsgDecrypt(String topic, MqttMessage mqttMessage) {
        String message = null;
        try {
            message = new String(mqttMessage.getPayload(), "UTF-8");
            boolean isPlain = StringUtils.contains(topic, "plain") || StringUtils.contains(topic, "PLAIN");
            if (!isPlain) {
                Coder coder = new Coder(EmqConstant.Config.GLOBAL_VERSION);
                byte[] payloadByte = coder.decrypt(mqttMessage.getPayload());
                message = new String(payloadByte, "UTF-8");
            }
        } catch (Exception e) {
            LOGGER.info("解析设备发布的消息异常->", e);
        }
        return message;
    }

    /**
     * 应答设备
     *
     * @param topic            主题
     * @param mqttBaseRequest  请求参数
     * @param mqttBaseResponse 响应参数
     */
    private void responseDevice(String topic, MqttBaseRequest mqttBaseRequest, MqttBaseResponse mqttBaseResponse) {
        mqttBaseResponse.setSrc(mqttBaseRequest.getDst());
        String src = mqttBaseRequest.getSrc();
        mqttBaseResponse.setDst(src);
        long seq = mqttBaseRequest.getSeq();
        mqttBaseResponse.setSeq(seq);
        mqttBaseResponse.setVersion(EmqConstant.Config.VERSION);
        mqttBaseResponse.setResult(EmqConstant.Code.SUCCESS);
        String method = mqttBaseRequest.getMethod();
        mqttBaseResponse.setMethod(method);
        if (!topic.contains(EmqConstant.Topic.PLAIN)) {
            DeviceMqttClient.getInstance().publishEncrpyMsg(mqttBaseResponse, src, seq);
        } else {
            DeviceMqttClient.getInstance().publishNoEncrpyMsg(mqttBaseResponse, src, seq);
        }
    }
}
